package cas.ihep.hadoop.da;

import cas.ihep.fs.UnifiedFileSystem;
import cas.ihep.hadoop.util.Utils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class YbjReducer5 extends Reducer<Text,Text,Text,Text> {
    private static final Log LOG = LogFactory.getLog(YbjReducer3.class);

    public void run(Reducer.Context context)throws IOException {
        try {
            Configuration conf = context.getConfiguration();
            UnifiedFileSystem ufs = UnifiedFileSystem.get(UnifiedFileSystem.newConfiguration(conf));
            String indir = conf.get(Utils.GlobalValueKey);
            if (indir != null && !indir.isEmpty()) {
                LOG.info("Deleting tmp global value: " + indir);
                ufs.delete(new URI(indir));
            }
            indir = conf.get(Utils.RunfileUsetmp);
            if (indir != null && !indir.isEmpty()) {
                indir = conf.get(Utils.RunfileKey);
                if (indir != null && !indir.isEmpty()) {
                    LOG.info("Deleting tmp runfile: " + indir);
                    ufs.delete(new URI(indir));
                }
            }
        }catch (URISyntaxException e){
            throw new IOException(e);
        }
    }
}
